package ex.other;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import java.util.concurrent.ExecutionException;

import static org.apache.flink.table.api.Expressions.$;


public class CommApiTestMysql {
    public static Table report(Table transactions) {
        throw new RuntimeException();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //        env.setParallelism(1);
        // 1     	 
    	
    	EnvironmentSettings settings = EnvironmentSettings
    		    .newInstance()
    		    .inStreamingMode()
    		    .build();

    		TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 2 
        String createInDDL = "CREATE TABLE inputTable (" +
                "username STRING, " +
                "url STRING, " +
                "ts BIGINT " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/input.txt'," +
                " 'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createInDDL);

        // 3  
        String createOutDDL = "CREATE TABLE spendreport (" +
                "username STRING, " +
                "url STRING, " +
                "ts BIGINT " +
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=UTC&useSSL=false'," +
                "   'table-name' = 'spendreport'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = 'liji@2024'," +
                "   'sink.buffer-flush.max-rows' = '0' "+
        ")";
        tableEnv.executeSql(createOutDDL);

        // 4 
        Table inputTable = tableEnv.from("inputTable");

        // 5 select user_name,ts as totalTs from clickTable where ts>500
        Table inputTableResult=inputTable
                .filter($("ts").isGreater(500))   ;

        // 6 

        tableEnv.createTemporaryView("result2", inputTableResult);
        Table resultTable2 = tableEnv.sqlQuery("select username,url,ts from result2");

        /**7.1-a 7.1-b   **/

        // 7.1-a 
        resultTable2.executeInsert("spendreport");
        // 7.1-b  
//        tableEnv.executeSql("insert into spendreport select *  from result2"); //

        /**  **/

       //7.2  
        String createPrintOutDDL = "CREATE TABLE printOutTable (" +
                "user_name STRING, " +
                "url STRING, " +
                "ts BIGINT"+
                ") WITH (" +
                " 'connector' = 'print' " +
                ")";
        tableEnv.executeSql(createPrintOutDDL);
        //
        resultTable2.executeInsert("printOutTable");

    }
}

